package com.example.demo.websocket;

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@ServerEndpoint(value = "/websocket/sumo")
@Slf4j
@Component
public class SumoWebsocket {
    /**
     * 全部在线会话  PS: 基于场景考虑 这里使用线程安全的Map存储会话对象。
     */
    private static Map<String, Session> onlineSessionMap = new ConcurrentHashMap<>();

    public static Map<String, Session> getOnlineSessionMap() {
        return onlineSessionMap;
    }

    @PostConstruct
    public void init() {
        System.out.println("kafka-测试----------------");
    }

    @OnOpen
    public void onOpen(Session session) {
        log.debug(session.getId() + "开启会话！");
        onlineSessionMap.put(session.getId(), session);
    }


    @OnMessage
    public void onMessage(Session session, String jsonStr) {
        log.info("received:{}", jsonStr);
    }


    @OnClose
    public void onClose(Session session) {
        onlineSessionMap.remove(session.getId());
    }

    /**
     * 服务端推送消息
     *
     * @param message
     */
    public synchronized void pushMessage(String message) {
        onlineSessionMap.entrySet().forEach(entry -> {
            Session session = entry.getValue();
            try {
                session.getBasicRemote().sendText(message);
            } catch (IOException e) {
                log.error(e.getMessage(), e);
            }
        });
    }

    /**
     * 当通信发生异常：打印错误日志
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error(error.getMessage(), error);
    }
}
